package mr.dividends;

import java.io.IOException;

import mr.dividends.model.DividendsKeyWritable;
import mr.dividends.model.DividendsValueWritable;

import org.apache.commons.lang.Validate;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DividendsJob extends Configured implements Tool {

	private static Logger LOG = LoggerFactory.getLogger(DividendsJob.class);
	public final static String PROP_INPUT_PATH = "input.path";
	public final static String PROP_OUTPUT_PATH = "output.path";


	@Override
	public int run(String[] arg0) throws Exception {
		Job job = Job.getInstance(getConf(), this.getClass().getName());
		job.setJarByClass(getClass());
		configureJob(job);
		return job.waitForCompletion(true) ? 0 : 1;
	}

	private void configureJob(Job job) throws IOException {
		// configure output and input source
		String input = getInput();

		DividendsInputFormat.addInputPath(job, new Path(input));
		job.setInputFormatClass(DividendsInputFormat.class);

		// configure mapper and reducer
		job.setMapperClass(DividendsMapper.class);
		job.setCombinerClass(DividendsCombiner.class);
		job.setReducerClass(DividendsReducer.class);

		// configure keys used between mappers and reducers
		job.setMapOutputKeyClass(DividendsKeyWritable.class);
		job.setMapOutputValueClass(DividendsValueWritable.class);

		// configure output
		Path outPath = getOutputPath();
		TextOutputFormat.setOutputPath(job, outPath);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
	}

	private String getInput() {
		String input = getConf().get(PROP_INPUT_PATH);
		Validate.notEmpty(input, "Must provide input path via ["
				+ PROP_INPUT_PATH + "] property");
		return input;
	}

	private Path getOutputPath() throws IOException {
		String out = getConf().get(PROP_OUTPUT_PATH);
		Validate.notEmpty(out, "You must provide value to find via ["
				+ PROP_OUTPUT_PATH + "]");
		Path outPath = new Path(out);
		FileSystem fs = FileSystem.get(getConf());
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
			LOG.info("Removed output directory [" + outPath + "]");
		}
		return outPath;
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		int exitCode = ToolRunner.run(new DividendsJob(), args);
		System.exit(exitCode);
	}

}
